package mq

import (
	"encoding/json"
	"fmt"
	"github.com/nsqio/go-nsq"
)

//nsqio消费测试

type MyTestHandler struct {
	q              *nsq.Consumer
	messageReceive int
}

// HandleMessage 消费者处理函数
func (h *MyTestHandler) HandleMessage(message *nsq.Message) error {
	type Data struct {
	}

	var (
		data *Data
		err  error
	)
	data = &Data{}
	if err = json.Unmarshal(message.Body, data); err != nil {
		/*todo 消费者处理消息逻辑*/
		fmt.Printf("Id:%s, Msg:%s \n", message.ID, string(message.Body))
		err = nil
	}
	message.Finish()
	return nil
}

func NewNsqConsumer(topic string, channel string, address string) {
	var (
		config *nsq.Config
		h      *MyTestHandler
		err    error
	)
	h = &MyTestHandler{}

	config = nsq.NewConfig()
	//注意MaxInFlight的设置，默认只能接受一个节点
	//config.MaxInFlight = 2

	if h.q, err = nsq.NewConsumer(topic, channel, config); err != nil {
		fmt.Println(err)
		return
	}

	h.q.AddHandler(h)
	//if err = h.q.ConnectToNSQD(address); err != nil {
	//	fmt.Println(err)
	//}
	if err = h.q.ConnectToNSQLookupd(address); err != nil {
		panic(err)
	}

	<-h.q.StopChan
	fmt.Println("stop")
	return
}
